package org.hbhk.hms.mq.tx.provider;

import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.hbhk.hms.mq.tx.mq.IPUtils;
import org.hbhk.hms.mq.tx.mq.JsonUtil;
import org.hbhk.hms.mq.tx.mq.MqMsgVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.Environment;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

public class DefaultMessageSender<T> implements MqMessageProvider<T> {

	protected static final Logger logger = LoggerFactory.getLogger(DefaultMessageSender.class);

	private IMqRecordService mqRecordService;

	private String namesrvAddr;

	private Environment environment;

	private MqTransactionManager mqTransactionManager;

	public static Map<String, DefaultMQProducer> mqProducerCache = Maps.newConcurrentMap();

	protected static Set<String> mqTxtabs = Sets.newHashSet();

	private SnowFlakeGenerator snowFlakeGenerator = new SnowFlakeGenerator();

	@Override
	public void send(MqMsgVo<T> msg) {

		if (msg == null) {
			throw new RuntimeException("mq msg  is empty");
		}
		T body = msg.getBody();
		if (body == null) {
			throw new RuntimeException("mq body  is empty");
		}
		Class<?> cls = AopTargetUtil.getProxyInterface(this);

		MqTxProducer mqTxProducer = cls.getAnnotation(MqTxProducer.class);

		String topic = mqTxProducer.topic();
		String namesrvAddr = mqTxProducer.namesrvAddr();

		String mqName = "defaultMQProducer";
		init();
		if (StringUtils.isEmpty(namesrvAddr)) {
			namesrvAddr = this.namesrvAddr;
		} else {
			mqName = cls.getName().replaceAll("\\.", "_");
		}
		String tags = msg.getTags();
		String txTable = mqTxProducer.txTable();
		mqTxtabs.add(txTable);
		String key = msg.getKey();
		if (StringUtils.isEmpty(key)) {
			throw new RuntimeException("msg key is empty");
		}
		String reqId = snowFlakeGenerator.nextId() + "";
		msg.put("msgId", reqId);
		msg.put("requestIp", IPUtils.getIpAddress());
		msg.put("msgKey", msg.getKey());
		MqRequest mqRequest = buildRpcRequest(JsonUtil.objectToJson(msg), topic, String.valueOf(key), mqName,
				namesrvAddr, txTable, tags, reqId);

		if (!TransactionSynchronizationManager.isSynchronizationActive()) {
			logger.debug("mq transaction is not active.");
			mqRecordService.sendMessage(mqRequest);
			return;
		}

		if (StringUtils.isEmpty(txTable)) {
			throw new RuntimeException("mq txTable is empty");
		}

		mqTransactionManager.execute(mqRequest);

	}

	private void init() {
		if (mqRecordService == null) {
			mqRecordService = MqTransactionManager.context.getBean(IMqRecordService.class);
			mqTransactionManager = MqTransactionManager.context.getBean(MqTransactionManager.class);
			environment = MqTransactionManager.context.getBean(Environment.class);
			namesrvAddr = resolve("${hms.mq.namesrvAddr}");
		}

	}

	private MqRequest buildRpcRequest(String body, String topic, String key, String mqName, String namesrvAddr,
			String txTable, String tags, String reqId) {
		buildDefaultMQProducer(mqName, namesrvAddr);
		MqRequest request = new MqRequest();
		request.setReqId(reqId);
		request.setBody(body);
		request.setMqName(mqName);
		request.setTopic(topic);
		request.setTxTable(txTable);
		request.setMsgKey(key);
		request.setTags(tags);
		return request;

	}

	private void buildDefaultMQProducer(String mqName, String namesrvAddr) {
		if (mqProducerCache.containsKey(mqName)) {
			return;
		}
		synchronized (logger) {
			if (!mqProducerCache.containsKey(mqName)) {
				String producerGroup = mqName + "hms-rocketmq-tx";
				DefaultMQProducer defaultMQProducer = new DefaultMQProducer(producerGroup);
				namesrvAddr = resolve(namesrvAddr);
				defaultMQProducer.setNamesrvAddr(namesrvAddr);
				try {
					defaultMQProducer.start();
				} catch (MQClientException e) {
					throw new RuntimeException(e);
				}
				mqProducerCache.put(mqName, defaultMQProducer);

			}
		}
	}

	private String resolve(String value) {
		if (org.springframework.util.StringUtils.hasText(value)) {
			return this.environment.resolvePlaceholders(value);
		}
		return value;
	}

}
